package com.shujia.advance;

import com.shujia.utils.BaseUtils;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;

/*
        1、通过手机号查询用户最近10条位置记录

        2、获取用户某一天在一个城市中的所有位置

     */
public class HbaseDianXin {
    public static void main(String[] args) throws Exception {
        Connection connection = BaseUtils.getConnection();
        Admin admin = connection.getAdmin();
        ResultScanner resultScanner = scanData(connection, "dianxin_data");
        printData(resultScanner);
    }
    public static void printData(ResultScanner resultScanner){
        //遍历结果得到每一行
        for (Result result : resultScanner) {
            StringBuilder sb = new StringBuilder();
            //处理每一行
            String[] strings = Bytes.toString(result.getRow()).split("_");
            sb.append("手机号：");
            String phoneNum = strings[0];
            sb.append(phoneNum).append("，进入时间：");
            long enterTime = 20180504000000L - Long.parseLong(strings[1]);
            sb.append(enterTime).append("，网格编号：");
            //手机号,网格编号,城市编号,区县编号,停留时间,进入时间,离开时间,时间分区
            String wg = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("wg")));
            sb.append(wg).append("，城市编号：");
            String cityId = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("cityId")));
            sb.append(cityId).append("，区县编号：");
            String qx = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("qx")));
            sb.append(qx).append("，停留时间：");
            String stayTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("stayTime")));
            sb.append(stayTime).append("，离开时间：");
            String leaveTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("leaveTime")));
            sb.append(leaveTime).append("，时间分区：");
            String dateTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("dateTime")));
            sb.append(dateTime);

            System.out.println(sb);
        }
    }

    public static ResultScanner scanData(Connection conn,String tableName) throws Exception{
        //获取表的实例对象
        Table table = conn.getTable(TableName.valueOf(tableName));

        Scan scan = new Scan();

        //创建行键前缀过滤器
        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("47BE1E866CFC071DB19D5E1C056BE28AE24C16E7"));
        //创建单列值过滤器
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"),Bytes.toBytes("cityId"), CompareOperator.EQUAL,Bytes.toBytes("83401"));

        FilterList filterList = new FilterList();
        filterList.addFilter(prefixFilter);
        filterList.addFilter(singleColumnValueFilter);
        scan.setFilter(filterList);

        return table.getScanner(scan);
    }


    public static void putsDianXianData(Connection conn,String pathName) throws Exception {
        //创建字符缓冲输入流对象
        BufferedReader br = new BufferedReader(new FileReader(pathName));
        ArrayList<Put> puts = new ArrayList<>();
        String line = null;
        //遍历每一行数据，将每一行中的每一列封装成Put集合
        while ((line = br.readLine()) != null) {
            String[] strings = line.split("\t");
            if (strings.length > 7 && !"\\N".equals(strings[1])) {
                String phoneName = strings[0];
                String wg = strings[1];
                String cityId = strings[2];
                String qx = strings[3];
                String stayTime = strings[4];
                String enterTime = strings[5]; // 20180503190539
                String leaveTime = strings[6];
                String dateTime = strings[7];

                //设计rowkey,为了需求查询最近的10条记录做准备（时间戳反转）
                //将手机号与减之后的enterTime拼接
                //20180503190539  20180504000000
                long time = 20180504000000L - Long.parseLong(enterTime);
                String rowKey = phoneName + "_" + time;

                //将每一列封装成Put对象
                Put put = new Put(Bytes.toBytes(rowKey));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("wg"),Bytes.toBytes(wg));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("cityId"),Bytes.toBytes(cityId));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("qx"),Bytes.toBytes(qx));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("stayTime"),Bytes.toBytes(stayTime));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("leaveTime"),Bytes.toBytes(leaveTime));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("dateTime"),Bytes.toBytes(dateTime));

                puts.add(put);
            }
        }

        //添加数据
        Table table = conn.getTable(TableName.valueOf("dianxin_data"));
        table.put(puts);
    }
}
